﻿#include "base/xrtc_pusher.h"

#include <sstream>

#include <rtc_base/task_utils/to_queued_task.h>
#include <rtc_base/logging.h>

#include "xrtc/ixrtc_engine.h"
#include "base/xrtc_utils.h"
#include "base/xrtc_global.h"
#include "base/xrtc_json.h"
#include "media_source/cam_source.h"

#include"websocket/ffwebsocketsignalclient.h"
#include<nlohmann/json.hpp>

using json = nlohmann::json;

namespace xrtc {

XRTCPusher::XRTCPusher() :
    current_thread_(rtc::Thread::Current()),
    websocket_client_(std::make_unique<FFWebSocketSignalClient>(this))
{
    //XRTCGlobal::Instance()->http_manager()->AddObject(this);
}

XRTCPusher::~XRTCPusher() {
    //XRTCGlobal::Instance()->http_manager()->RemoveObject(this);
    Stop();
}

void XRTCPusher::OnAnswerReady(WebRTCStream* stream, 
    const std::string& answer) 
{
    current_thread_->PostTask(webrtc::ToQueuedTask([=]() {
        if (request_params_["uid"].empty() || request_params_["streamName"].empty()) {
            RTC_LOG(LS_WARNING) << "send answer failed, invalid url: " << url_;
            return;
        }

        // 构造sendanswer的请求
        std::string url = "https://" + host_ + "/signaling/sendanswer";
        std::stringstream body;
        body << "uid=" << request_params_["uid"]
            << "&streamName=" << request_params_["streamName"]
            << "&type=push"
            << "&answer=" << HttpManager::UrlEncode(answer);

        HttpRequest request(url, body.str());

        // 发送信令请求
        XRTCGlobal::Instance()->http_manager()->Post(request,
            [=](HttpReply reply) {
                RTC_LOG(LS_INFO) << "signaling sendanswer response, "
                    << "url: " << reply.get_url()
                    << ", body: " << reply.get_body()
                    << ", status: " << reply.get_status_code()
                    << ", err_no: " << reply.get_errno()
                    << ", err_msg: " << reply.get_err_msg()
                    << ", response: " << reply.get_resp();


                if (!ParseReply(reply)) {
                    SendStartPushEvent(XRTCError::kStartPushSendAnswerErr);
                    return;
                }

            }, this);
    }));
}

void XRTCPusher::OnStreamState(WebRTCStream* stream, 
    XRTCStreamState state) 
{
    current_thread_->PostTask(webrtc::ToQueuedTask([=]() {
        XRTCEngineObserver* observer = XRTCGlobal::Instance()->engine_observer();
        if (observer) {
            observer->OnPushStreamState(this, state);
        }
    }));
}

void XRTCPusher::StartRender(webrtc::VideoTrackInterface* video_track)
{
    if (!canvas_.view) {
        return;
    }

    render_ = std::make_unique<D3D9Render>(canvas_.view,video_track);
    render_->Start();

}

bool XRTCPusher::ParseReply(const HttpReply& reply, 
    std::string& type, 
    std::string& sdp) 
{
    if (reply.get_status_code() != 200 || reply.get_errno() != 0) {
        RTC_LOG(LS_WARNING) << "http response error, status: "
            << reply.get_status_code()
            << ", err_no: " << reply.get_errno();
        return false;
    }

    JsonValue value;
    if (!value.FromJson(reply.get_resp())) {
        RTC_LOG(LS_WARNING) << "invalid json response";
        return false;
    }

    JsonObject jobj = value.ToObject();
    int err_no = jobj["errNo"].ToInt();
    if (err_no != 0) {
        RTC_LOG(LS_WARNING) << "http response errNo is not 0, errNo: " << err_no;
        return false;
    }

    JsonObject jdata = jobj["data"].ToObject();
    type = jdata["type"].ToString();
    sdp = jdata["sdp"].ToString();

    if (sdp.empty()) {
        RTC_LOG(LS_WARNING) << "offer sdp is empty";
        return false;
    }

    return true;
}

bool XRTCPusher::ParseReply(const HttpReply& reply) {
    if (reply.get_status_code() != 200 || reply.get_errno() != 0) {
        RTC_LOG(LS_WARNING) << "http response error, status: "
            << reply.get_status_code()
            << ", err_no: " << reply.get_errno();
        return false;
    }

    JsonValue value;
    if (!value.FromJson(reply.get_resp())) {
        RTC_LOG(LS_WARNING) << "invalid json response";
        return false;
    }

    JsonObject jobj = value.ToObject();
    int err_no = jobj["errNo"].ToInt();
    if (err_no != 0) {
        RTC_LOG(LS_WARNING) << "http response errNo is not 0, errNo: " << err_no;
        return false;
    }

    return true;
}

void XRTCPusher::DoOffer(std::string const offer)
{
    json JMsg;
    JMsg["cmd"] = SIGNAL_TYPE_OFFER;
    JMsg["roomId"] = roomId_;
    JMsg["uid"] = uid_;
    JMsg["remoteUid"] = remoteUid_;

    // 关键：构造与前端一致的 SDP 对象（包含 type 和 sdp）
    json sdp_obj;
    sdp_obj["type"] = "offer";  // 对应前端 session.type
    sdp_obj["sdp"] = offer;     // 对应前端 session.sdp（SDP 原始字符串）

    // 将 sdp_obj 序列化为字符串，作为 message 字段的值（与前端 JSON.stringify(session) 一致）
    JMsg["message"] = sdp_obj.dump();

    Send(JMsg.dump());
}

void XRTCPusher::DoAnswer(std::string const answer)
{
    json JMsg;
    JMsg["cmd"] = SIGNAL_TYPE_ANSWER;
    JMsg["roomId"] = roomId_;
    JMsg["uid"] = uid_;
    JMsg["remoteUid"] = remoteUid_;

    // 关键：构造与前端一致的 SDP 对象（包含 type 和 sdp）
    json sdp_obj;
    sdp_obj["type"] = "answer";  // 对应前端 session.type
    sdp_obj["sdp"] = answer;     // 对应前端 session.sdp（SDP 原始字符串）

    // 将 sdp_obj 序列化为字符串，作为 message 字段的值（与前端 JSON.stringify(session) 一致）
    JMsg["message"] = sdp_obj.dump();

    Send(JMsg.dump());
}

void XRTCPusher::SendStartPushEvent(XRTCError err) {
    XRTCEngineObserver* observer =
        XRTCGlobal::Instance()->engine_observer();
    if (observer) {
        observer->OnStartPushEvent(this, err);
    }
}

void XRTCPusher::SendStopPushEvent(XRTCError err) {
    XRTCEngineObserver* observer = XRTCGlobal::Instance()->engine_observer();
    if (observer) {
        observer->OnStopPushEvent(this, err);
    }
}

bool XRTCPusher::StartPush(const std::string& url, 
    IXRTCVideoSource* video_source, const XRTCCanvas& canvas,std::string uid,std::string roomId)
{
    current_thread_->PostTask(webrtc::ToQueuedTask([=]() {

#if 0
        // 解析推流地址
        // xrtc://www.str2num.com:8021/push?uid=xxx&streamName=xxx
        if (!ParseXRTCUrl(url, protocol_, host_, action_, request_params_)) {
            return;
        }
        // 构造信令请求
        std::string url = "https://" + host_ + "/signaling/push";
        std::stringstream body;
        body << "uid=" << request_params_["uid"]
            << "&streamName=" << request_params_["streamName"]
            << "&audio=" << request_params_["audio"]
            << "&video=" << request_params_["video"]
            << "&isDtls=" << request_params_["isDtls"];

        HttpRequest request(url, body.str());

        // 发送信令请求
        XRTCGlobal::Instance()->http_manager()->Post(request,
            [=](HttpReply reply) {
                RTC_LOG(LS_INFO) << "signaling push response, "
                    << "url: " << reply.get_url()
                    << ", body: " << reply.get_body()
                    << ", status: " << reply.get_status_code()
                    << ", err_no: " << reply.get_errno()
                    << ", err_msg: " << reply.get_err_msg()
                    << ", response: " << reply.get_resp();

                std::string type;
                std::string sdp;
                
                if (!ParseReply(reply, type, sdp)) {
                    SendStartPushEvent(XRTCError::kStartPushRequestOfferErr);
                    return;
                }

                RTC_LOG(LS_INFO) << "offer sdp: \n" << sdp;

                uint64_t uid = std::atoll(request_params_["uid"].c_str());
                std::string stream_name = request_params_["stream_name"];

                // 创建PushStream
                push_stream_ = std::make_unique<PushStream>(this,std::to_string(uid), stream_name);
                push_stream_->SignalAnswerReady.connect(this,
                    &XRTCPusher::OnAnswerReady);
                push_stream_->SignalStreamState.connect(this,
                    &XRTCPusher::OnStreamState);
                push_stream_->Init();
                push_stream_->AddAudioTrack();
                push_stream_->AddVideoTrack(CamSource::Create(
                    dynamic_cast<rtc::VideoSourceInterface<webrtc::VideoFrame>*>(video_source)));
                push_stream_->SetRemoteOffer(sdp);

            }, this);
#endif 

        video_source_ = video_source;
        canvas_ = canvas;
        url_ = url;

        uid_ = uid;
        roomId_ = roomId;

        //启动事件循环、连接服务器
        Start(url_);
        //加入房间
        DoJoin(roomId_, uid_);

    }));

    return true;
}

void XRTCPusher::Start(std::string const url) {
    if (running_) {
        RTC_LOG(WARNING) << "XRTCPusher Already Start";
        return;
    }

    // 启动WebSocket线程
    running_ = true;
    websocket_thread_ = new std::thread(
        [=]()
        {
            rtc::SetCurrentThreadName("websocket_thread");

            {
                //初始化WebSocket
                std::lock_guard<std::mutex>lock(mutex_);
                websocket_client_->init();
                websocket_client_->connect(url);
            }

            do {
                //启动事件循环
                RTC_LOG(LS_INFO) << "XRTCPusher Start";
                websocket_client_->exec();
            } while (running_);

            RTC_LOG(LS_INFO) << "XRTCPusher Stop";
        });
}

void XRTCPusher::Stop()
{
    if (!running_) {
        RTC_LOG(WARNING) << "XRTCPusher Already Stop";
        return;
    }
    running_ = false;

    //停止事件循环
    std::lock_guard<std::mutex> lock(mutex_);

    int ret = websocket_client_->close();
    if (ret != 0) {
        RTC_LOG(LS_WARNING) << "XRTCPusher close error";
    }
    else {
        RTC_LOG(LS_INFO) << "XRTCPusher close success";
    }

    //等待线程退出
    if (websocket_thread_ && websocket_thread_->joinable()) {
        websocket_thread_->join();
        RTC_LOG(LS_INFO) << "XRTCPusher Thread Join";

        delete websocket_thread_;
        websocket_thread_ = nullptr;

    }

    RTC_LOG(LS_INFO) << "XRTCPusher Stop";
}

void XRTCPusher::StopPush() {
    current_thread_->PostTask(webrtc::ToQueuedTask([=]() {
#if 0
        push_stream_.reset();

        // 构造stoppush的请求
        std::string url = "https://" + host_ + "/signaling/stoppush";
        std::stringstream body;
        body << "uid=" << request_params_["uid"]
            << "&streamName=" << request_params_["streamName"];

        HttpRequest request(url, body.str());

        // 发送信令请求
        XRTCGlobal::Instance()->http_manager()->Post(request,
            [=](HttpReply reply) {
                RTC_LOG(LS_INFO) << "signaling stoppush response, "
                    << "url: " << reply.get_url()
                    << ", body: " << reply.get_body()
                    << ", status: " << reply.get_status_code()
                    << ", err_no: " << reply.get_errno()
                    << ", err_msg: " << reply.get_err_msg()
                    << ", response: " << reply.get_resp();

                XRTCError err = XRTCError::kNoErr;
                bool res = ParseReply(reply);
                if (!res) {
                    err = XRTCError::kStopPushRequestError;
                }

                SendStopPushEvent(err);


            }, this);
#endif 
 
        push_stream_.reset();
        DoLeave();
    }));
}



//发送candidate到对端
void XRTCPusher::DoCandidate(const webrtc::IceCandidateInterface* candidate)
{

    std::string candidate_str;
    candidate->ToString(&candidate_str); // 获取原始 candidate 字符串（如"candidate:xxx..."）
    std::string sdp_mid = candidate->sdp_mid(); // 获取 sdpMid（如"0"）
    int sdp_mline_index = candidate->sdp_mline_index(); // 获取 sdpMLineIndex（如0）

    // 构造内层消息JSON
    json candidate_obj;
    candidate_obj["candidate"] = candidate_str;
    candidate_obj["sdpMid"] = sdp_mid;
    candidate_obj["sdpMLineIndex"] = sdp_mline_index;

    //构造外层消息 JSON
    json JMsg;
    JMsg["cmd"] = SIGNAL_TYPE_CANDIDATE;
    JMsg["roomId"] = roomId_;
    JMsg["uid"] = uid_;
    JMsg["remoteUid"] = remoteUid_;
    JMsg["message"] = candidate_obj.dump(); // 序列化内层对象为字符串（与前端 JSON.stringify 一致）

    Send(JMsg.dump());
    RTC_LOG(LS_INFO) << "Do Candidate :\n" << JMsg.dump();
}

//加入房间
void XRTCPusher::DoJoin(std::string const roomId, std::string const uid)
{
    current_thread_->PostTask(RTC_FROM_HERE, [=](){
        uid_ = uid;
        roomId_ = roomId;

        json JMsg;
        JMsg.emplace("cmd", xrtc::SIGNAL_TYPE_JOIN);
        JMsg.emplace("uid", uid_);
        JMsg.emplace("roomId", roomId_);

        Send(JMsg.dump());
        });

}

//离开房间
void XRTCPusher::DoLeave()
{
    current_thread_->PostTask(RTC_FROM_HERE, [=]() {
        json JMsg;
        JMsg.emplace("cmd", xrtc::SIGNAL_TYPE_LEAVE);
        JMsg.emplace("uid", uid_);
        JMsg.emplace("roomId", roomId_);

        Send(JMsg.dump());
        SendStopPushEvent(XRTCError::kNoErr);
    });
}

void XRTCPusher::DoRender(WebRTCStream* stream, rtc::scoped_refptr<webrtc::RtpReceiverInterface> receiver)
{
    webrtc::MediaStreamTrackInterface* track = receiver->track().release();
    if (track->kind() == webrtc::MediaStreamTrackInterface::kVideoKind) {
        webrtc::VideoTrackInterface* video_track = static_cast<webrtc::VideoTrackInterface*>(track);
        StartRender(video_track);
    }
}

int XRTCPusher::Send(std::string const message) {
    if (!running_) {
        RTC_LOG(WARNING) << "XRTCPusher Not Init";
        SendStartPushEvent(XRTCError::kStartPushSocketError);
        return -1;
    }

    //发送消息
    std::unique_lock<std::mutex> auto_lock(mutex_);
    int ret = websocket_client_->send(message);

    //尝试重发
    if (ret == WEBSOCKET_AGAIN) {
        size_t reTry_ = 0;
        do {
            auto_lock.unlock(); //释放锁
            Sleep(100); //100ms休眠
            auto_lock.lock();

            ret = websocket_client_->send(message);
            reTry_++;
        } while (ret == WEBSOCKET_AGAIN && reTry_ < 10);

        //判断是否发送成功
        if (ret == WEBSOCKET_SUCESS) {
            RTC_LOG(LS_INFO) << "XRTCPusher Send :" << message;
            SendStartPushEvent(XRTCError::kNoErr);
        }
        else {
            RTC_LOG(LS_WARNING) << "XRTCPusher Send Fail!";
            SendStartPushEvent(XRTCError::kStartPushSocketError);
        }
    }

    return ret;
}


void XRTCPusher::DoStreamStateChange(WebRTCStream* stream, XRTCStreamState state)
{
  /*  current_thread_->Invoke<void>(RTC_FROM_HERE, [=]() {
        XRTCEngineObserver* observer = XRTCGlobal::Instance()->engine_observer();
        if (observer) {
            observer->OnPushStreamStateEvent(this, state);
        }
        });*/

}

//停止推流
void XRTCPusher::DoStopStream(xrtc::XRTCError err)
{
   /* XRTCEngineObserver* observer = XRTCGlobal::Instance()->engine_observer();
    if (observer) {
        observer->OnStopStreamEvent(this, err);
    }*/

}


//对端加入
void XRTCPusher::handleRemoteNewPeer(const json JMsg)
{
    current_thread_->PostTask(RTC_FROM_HERE, [=]() {
        //获取sdp
        try {
            remoteUid_ = JMsg.at("remoteUid");
          
            // 创建PushStream
            push_stream_ = std::make_unique<PushStream>(this, uid_, roomId_);
            push_stream_->SignalAnswerReady.connect(this,&XRTCPusher::OnAnswerReady);
            push_stream_->SignalStreamState.connect(this,&XRTCPusher::OnStreamState);

            push_stream_->Init();
            push_stream_->AddVideoTrack(CamSource::Create(
                dynamic_cast<rtc::VideoSourceInterface<webrtc::VideoFrame>*>(video_source_)));
            push_stream_->AddAudioTrack();

            //保存对端Offer
            push_stream_->CreateOffer();

        }
        catch (std::exception const e) {
            std::cerr << "Parse JSON Failed!" << std::endl;
        }
    });

}

//本地加入
void XRTCPusher::handleResponseJoin(const json JMsg)
{
    current_thread_->PostTask(RTC_FROM_HERE, [=]() {
        try {
            remoteUid_ = JMsg.at("remoteUid");
        }
        catch (std::exception const e) {
            std::cout << "Parse JSON Failed!" << std::endl;
        }
    });

}

//对端离开
void XRTCPusher::handleRemotePeerLeave(const json JMsg)
{
    current_thread_->PostTask(RTC_FROM_HERE, [=]() {
        remoteUid_.clear();
        XRTCEngineObserver* observer = XRTCGlobal::Instance()->engine_observer();
        if (observer) {
            observer->OnRemotePeerLeaveEvent();
        }
    });
}

//收到offer
void XRTCPusher::handleRemoteOffer(const json JMsg)
{
    current_thread_->PostTask(RTC_FROM_HERE, [=]() {
        //获取sdp
        try {
            std::string msg = JMsg.at("message");

            // //msg也是JSON格式的，需要解析d
            json json_sdp = json::parse(msg);
            std::string sdp = json_sdp.at("sdp");

            // 创建PushStream
            push_stream_ = std::make_unique<PushStream>(this, uid_, roomId_);
            push_stream_->SignalAnswerReady.connect(this,
                &XRTCPusher::OnAnswerReady);
            push_stream_->SignalStreamState.connect(this,
                &XRTCPusher::OnStreamState);

            push_stream_->Init();
          /*  std::unique_ptr<rtc::VideoSourceInterface<webrtc::VideoFrame>> cam_source_ptr =
                absl::WrapUnique<rtc::VideoSourceInterface<webrtc::VideoFrame>>(
                    dynamic_cast<rtc::VideoSourceInterface<webrtc::VideoFrame>*>(video_source_));

            push_stream_->AddVideoTrack(CamSource::Create(std::move(cam_source_ptr)));*/
            push_stream_->AddVideoTrack(CamSource::Create(
                dynamic_cast<rtc::VideoSourceInterface<webrtc::VideoFrame>*>(video_source_)));
            push_stream_->AddAudioTrack();

            //保存对端Offer
            RTC_LOG(LS_INFO) << "remote offer sdp \n" << sdp;
            push_stream_->SetRemoteOffer(sdp);

        }
        catch (std::exception const e) {
            std::cerr << "Parse JSON Failed!" << std::endl;
        }
    });
}

//收到answer
void XRTCPusher::handleRemoteAnswer(const json JMsg)
{
    current_thread_->PostTask(RTC_FROM_HERE, [=]() {
        //获取sdp
        try {
            std::string msg = JMsg.at("message");

            //msg也是JSON格式的，需要解析d
            json json_sdp = json::parse(msg);
            std::string sdp = json_sdp.at("sdp");
          
            //保存对端Offer
            RTC_LOG(LS_INFO) << "remote answer sdp \n" << sdp;
            push_stream_->SetRemoteAnswer(sdp);
        }
        catch (std::exception const e) {
            std::cerr << "Parse JSON Failed!" << std::endl;
        }
        });

}

//收到candidate，添加到本地
void XRTCPusher::handleRemoteCandidate(const json JMsg)
{
    current_thread_->Invoke<void>(RTC_FROM_HERE, [=]() {
        //获取sdp
        try {
            std::string msg = JMsg.at("message");

            //msg也是JSON格式的，需要解析
            json json_sdp = json::parse(msg);

            //存储sdp_mid,sdp_mlineindex,sdp
            std::string sdp_mid = json_sdp.at("sdpMid");
            int sdp_mlineindex = json_sdp.at("sdpMLineIndex");
            std::string sdp = json_sdp.at("candidate");

            push_stream_->AddIceCandidate(sdp_mid, sdp_mlineindex, sdp);
        }
        catch (std::exception const e) {
            std::cerr << "Parse JSON Failed!" << std::endl;
        }
        });

}


} // namespace xrtc